package org.ala.tiktools.rpc.kcp.netty.server.handler;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.util.List;

import org.ala.tiktools.logs.LoggerFactory;
import org.ala.tiktools.rpc.kcp.config.RpcKcpConfigInfo;
import org.ala.tiktools.rpc.kcp.netty.IKcpServerChannelConfig;
import org.ala.tiktools.rpc.kcp.netty.KcpPacket;
import org.ala.tiktools.rpc.kcp.netty.NettyKcp;
import org.ala.tiktools.rpc.kcp.netty.server.KcpServerChannel;
import org.ala.tiktools.rpc.kcp.netty.server.KcpServerChildChannel;
import org.ala.tiktools.rpc.kcp.netty.tools.CodecOutputList;
import org.ala.tiktools.rpc.kcp.netty.tools.NettyTools;
import org.ala.tiktools.tools.Assert;
import org.ala.tiktools.tools.TimestampTools;
import org.slf4j.Logger;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.RecvByteBufAllocator;
import io.netty.util.internal.PlatformDependent;


/**
 * ServerChannel负责的Reader
 *
 * @author ala
 * @date 2025-04-15 16:59
 */
public class KcpServerChannelReader {
    static Logger log = LoggerFactory.getLogger("rpc", "server");
    /**
     *
     */
    protected KcpServerChannel serverChannel;
    public KcpServerChannelReader(KcpServerChannel serverChannel) {
        this.serverChannel = serverChannel;
    }


    /**
     *  接收数据
     *  <p> 从udp socket中取数
     *  <p> 该过程只能在boss线程内（boss线程负责 udp socket 的epoll）
     */
    @SuppressWarnings("deprecation")
	public int doReadMessages(List<Object> datas) throws Exception {
        //  拿真实的 udp channel
        DatagramChannel ch = (DatagramChannel) serverChannel.javaChannel();
        IKcpServerChannelConfig config = serverChannel.config();
        RecvByteBufAllocator.Handle handle = serverChannel.unsafe().recvBufAllocHandle();

        //  分配一个byteBuf，用于接收数据
        ByteBuf buf = handle.allocate(config.getAllocator());
        //  预估本次接受的数据量大小（还没有实际读取，这里直接取的data的可写区域。但实际可能读的比这少）
        handle.attemptedBytesRead(buf.writableBytes());
        boolean free = true;
        try {
            //  返回底层的ByteBuffer视图（buf的可写区）
            ByteBuffer buffer = buf.internalNioBuffer(buf.writerIndex(), buf.writableBytes());
            int pos = buffer.position();
            InetSocketAddress remoteAddr = (InetSocketAddress) ch.receive(buffer);
            if (remoteAddr == null) { return 0; }

            //  通过写针变化算出读了多少数据
            handle.lastBytesRead(buffer.position() - pos);
            //  buf的写针右移量 = buffer的读入数据量
            int len = handle.lastBytesRead();
            datas.add(KcpPacket.newInstance(buf.writerIndex(buf.writerIndex() + len), remoteAddr));
            free = false;
//            log.info("[KcpServerChannelReader doReadMessages] 服务端收到消息: ({})字节", len);
            return len;
        } catch (Throwable e) {
            PlatformDependent.throwException(e);
            return -1;
        } finally {
            //  如果读取失败，刚刚分配出来的ByteBuf要释放掉
            if (free) {
                buf.release();
            }
        }
    }


    /**
     *  unsafe read方法做的事情
     *  <p> boss线程版本
     */
    @SuppressWarnings("deprecation")
	public void unsafeReadInBoss(List<Object> readBuf, RecvByteBufAllocator.Handle alloctorHandler) {
        Assert.assertTrue(serverChannel.eventLoop().inEventLoop(), "read方法的原始读取只允许在BossEventLoop线程中执行");
        IKcpServerChannelConfig config = serverChannel.config();
        ChannelPipeline pipeline = serverChannel.pipeline();
        alloctorHandler.reset(config);
        //  读取消息进readBuf。读取出来都是KcpPacket
        boolean closed = false;
        Throwable ex = null;
        try {
            do {
                int len = doReadMessages(readBuf);
                if (len == 0) { break; }
                if (len < 0) { closed = true; break; }
            } while (alloctorHandler.continueReading());
        } catch (Exception e) {
            ex = e;
        }
        //  遍历读到的KcpPacket
        for (Object o : readBuf) {
            KcpPacket kp = (KcpPacket) o;
            InetSocketAddress addr = kp.getRemoteAddress();
            ByteBuf data = kp.getContent();
            //  如果kp的channel已经在close列表中，把rcv里面的数据读出来，再release掉（已经关掉的channel不应该再继续消费的）
            KcpServerChannelCloser.CloseWaitTask task = serverChannel.getCloser().getCloseWaitTask(addr);
            CodecOutputList<ByteBuf> list = null;
            Exception subEx = null;
            if (task != null) {
                NettyKcp nkcp = task.nkcp;
                //  读取数据，并不断重读recv
                try {
                    nkcp.input(data);
                    //  反复的把消息读出来，然后销毁
                    while (nkcp.canRecv()) {
                        if (list == null) { list = CodecOutputList.newInstance(); }
                        nkcp.receive(list);
                    }
                } catch (IOException e) {
                    subEx = e;
                } finally {
                    kp.recycle();
                }
                //  如果读到了数据，全部销毁
                if (list != null) {
                    clearAndRelease(list);
                }
                //  如果input的过程中出错了，当场关闭
                if (subEx != null) {
                    serverChannel.getCloser().removeCloseWaitTask(addr);
                    nkcp.setKcpClosed();
                }
            }
            //  如果不在关闭列表中，处理该消息
            else {
                KcpServerChildChannel childChannel = serverChannel.getCreator().getOrCreateKcpChannel(addr);
                if (!childChannel.isActive()) {
                    kp.release();
                    continue;
                }
                ChannelPipeline childPipeline = childChannel.pipeline();
                boolean mergeSegmentBuf = RpcKcpConfigInfo.INSTANCE.isIkcpMergeSegment();
                CodecOutputList<ByteBuf> recvBufList = null;
                try {
                    //  把data中的数据读进recvBuf，设置当前时间update
                	//	TODO	这里其实有问题，并没有处理拆包问题，每个数据包内必须包含完整kcp数据包
                	//			tcp是新弄了一个ByteBuf，接收读过来的字节流，再用新弄的ByteBuf解码。不过那样就多了一次复制（详见ByteToMessageDecode .Cumulator cumulator = MERGE_CUMULATOR;）
                	//			kcp的做法其实是带宽换时间，增加网络请求数量，减少数据复制
                    childChannel.kcpInput(data);
                    //  设置update时间戳，把积压的消息flush出去
                    childChannel.kcpTsUpdate(TimestampTools.milliSeconds32());

                    //  如果合并消息，每读到一个完整消息，触发一次fireChannelRead，直到读不到完整消息为止
                    if (mergeSegmentBuf) {
                        ByteBufAllocator allocator = config.getAllocator();
                        int peekSize = 0;
                        while ((peekSize = childChannel.peekSize()) >= 0) {
                            ByteBuf recvBuf = allocator.ioBuffer(peekSize);
                            childChannel.kcpReceive(recvBuf);
                            childPipeline.fireChannelRead(recvBuf);
                            childPipeline.fireChannelReadComplete();
                        }
                    }
                    //  如果不合并，每个数据包触发一次fireChannelRead。但需要自行处理拆包
                    else {
                        while (childChannel.kcpCanRecv()) {
                        	if (recvBufList == null) { recvBufList = CodecOutputList.newInstance(); }
                            childChannel.kcpReceive(recvBufList);
                        }
                        if (recvBufList != null) {
                        	NettyTools.fireChannelRead(childChannel, recvBufList);
                            recvBufList.recycle();
                        }
                    }
                } catch (Exception e) {
                    subEx = e;
                } finally {
                    kp.release();
                }
                //  如果读的时候发生了异常
                if (subEx != null) {
                    NettyTools.fireExceptionAndClose(childChannel, subEx);
                }
            }
        }
        //  释放readBuf，如果需要关闭，执行关闭
        readBuf.clear();
        alloctorHandler.readComplete();
        //  如果出现异常
        if (ex != null) {
            closed = serverChannel.closeOnReadError(ex);
            pipeline.fireExceptionCaught(ex);
        }
        //  如果确实需要关闭
        if (closed) {
            if (serverChannel.isOpen()) {
                serverChannel.close(serverChannel.voidPromise());
            }
        }
    }
    /**
     *  销毁未读的消息
     */
    private void clearAndRelease(CodecOutputList<ByteBuf> list) {
        if (list == null) {return;}
        for (int i = 0 ; i < list.size() ; i++) {
            list.getUnSafe(i).release();
        }
        list.recycle();
    }

}
